package com.bw.gamll.realtime.dwd.db.app;

import com.bw.gmall.realtime.common.base.BaseSqlApp;
import com.bw.gmall.realtime.common.constant.Constant;
import com.bw.gmall.realtime.common.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @projectName: gmalll2206
 * @package: com.bw.gamll.realtime.dwd.db.app
 * @className: DwdTradeCartAdd
 * @author: DuanYu
 * @description: TODO
 * @date: 2025/2/25 19:49
 * @version: 1.0
 */

public class DwdTradeCartAdd extends BaseSqlApp {
    public static void main(String[] args) {
        new DwdTradeCartAdd().start(Constant.TOPIC_DB, Constant.TOPIC_DWD_TRADE_CART_ADD, 4, 10013);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv, String groupId) {
        readOdsDb(tableEnv, Constant.TOPIC_DWD_TRADE_CART_ADD);
        Table cart_info = tableEnv.sqlQuery("select `data`['id'] id, " +
                "`data`['user_id'] user_id, " +
                "`data`['sku_id'] sku_id, " +
                "cast(`data`['cart_price'] as double) cart_price, " +
                "if (`type` = 'insert',cast(`data`['sku_num'] as int),cast(`data`['sku_num'] as int) - cast(`old`['sku_num'] as int )) sku_num, " +
                "ts " +
                "from topic_db " +
                "where `database`='gmall' " +
                "and `table`='cart_info' " +
                "and (" +
                " `type`='insert' " +
                "  or(" +
                "     `type`='update' " +
                "      and `old`['sku_num'] is not null " +
                "      and cast(`data`['sku_num'] as int) > cast(`old`['sku_num'] as int) " +
                "   )" +
                ")");
        tableEnv.createTemporaryView("cart_info", cart_info);
//        tableEnv.sqlQuery("select * from cart_info limit 5").execute().print();
        tableEnv.executeSql("create table dwd_trade_cart_add(" +
                "   id string, " +
                "   user_id string," +
                "   sku_id string," +
                "   cart_price decimal(10,2)," +
                "   sku_num int, " +
                "   ts  bigint " +
                ")" + SQLUtil.getKafkaSinkSQL(Constant.TOPIC_DWD_TRADE_CART_ADD));

        cart_info.executeInsert(Constant.TOPIC_DWD_TRADE_CART_ADD);
    }

}
